package cn.itcast.szse;

import cn.itcast.avro.SzseAvro;
import cn.itcast.kafka.KafkaPro;

import java.io.*;
import java.math.BigDecimal;
import java.net.Socket;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;

//解析文件获取成交量和成交金额数据，并封装到Map<String, Map<String, Long>> map中。
public class SocketClient {
    private static Map<String, Map<String, Long>> map = new HashMap<>();

    //随机浮动成交价格系数
    private static Double[] price = new Double[]{0.1, 0.11, 0.12, 0.13, 0.14, 0.15, 0.16, 0.17, 0.18, 0.19, 0.2, -0.1, -0.11, -0.12, -0.13, -0.14, -0.15, -0.16, -0.17, -0.18, -0.19, -0.2};

    //随机浮动成交量
    private static int[] volumn = new int[]{50, 80, 110, 140, 170, 200, 230, 260, 290, 320, 350, 380, 400, 500, 600, 700, 800, 900, 1000, 1100, 1200, 1300};

    /**
     * 开发步骤：
     * 1.读取本地文件（深市）
     * 2.解析行数据获取成交量和成交金额
     * 3.封装数据到map
     * 4.关流
     */
    public static void main(String[] args) throws Exception {

        //2.建立socket连接，获取流数据
        final Socket socket = new Socket("localhost", 4444);
        final InputStream inputStream = socket.getInputStream();
        final DataInputStream dataInputStream = new DataInputStream(inputStream);

        final KafkaPro<SzseAvro> kafkaPro = new KafkaPro<>();
        //3.读文件缓存成交量和成交金额
        parseFile();

        while (true){
            final String readUTF = dataInputStream.readUTF();
            SzseAvro szseAvro = tranform(readUTF);

            kafkaPro.sendData("szse",szseAvro);
//            System.out.println(szseAvro);
        }


/*        for (Map.Entry<String, Map<String, Long>> entry : map.entrySet()) {
            System.out.println(entry.getKey()+entry.getValue());
        }*/
    }

    private static SzseAvro tranform(String readUTF) {

        /**
         * 开发步骤：
         * 1.拷贝成交量和成交价数组
         * 2.获取随机浮动成交量和价格
         * 3.字符串切割、计算最新价
         * 4.获取缓存的成交量/金额
         * 5.计算总成交量/金额
         * 6.缓存总成交量/金额
         * 7.获取最高价和最低价(和最新价比较)
         * 8.封装结果数据
         */
        final String[] arr = readUTF.split("\\|");

        //2.获取随机浮动成交量和价格
        Random random = new Random();
        Double priceRandom = price[random.nextInt(price.length)];
        final int volumnRandom = volumn[random.nextInt(volumn.length)];

        final String code = arr[1].trim();


        //3.计算最新价
        final BigDecimal tradePriceBase = new BigDecimal(arr[9].trim());
        final BigDecimal tradePriceNew = tradePriceBase.multiply(new BigDecimal(1 + priceRandom)).setScale(2, BigDecimal.ROUND_HALF_UP);

        //4.获取缓存的成交量/金额
        final Map<String, Long> volAmtMap = map.get(code);
        final Long tradeVolCache = volAmtMap.getOrDefault("tradeVol", 0L);
        final Long tradeAmtCache = volAmtMap.getOrDefault("tradeAmt", 0L);

        //总成交量
        Long tradeVolTotal =tradeVolCache+volumnRandom;

        final BigDecimal tradeAmtNew = tradePriceNew.multiply(new BigDecimal(volumnRandom)).setScale(2, BigDecimal.ROUND_HALF_UP);
        //总成交金额
        Long tradeAmtTotal = tradeAmtCache+ tradeAmtNew.longValue();
        //5.缓存最新成交量和成交金额
        volAmtMap.put("tradeVol",tradeVolTotal);
        volAmtMap.put("tradeAmt",tradeAmtTotal);
        map.put(code,volAmtMap);

        //6.获取最高价和最低价
        //计算最高价
        BigDecimal highPrice = new BigDecimal(arr[7].trim());

        if (tradePriceNew.compareTo(highPrice)==1){
            highPrice=tradePriceNew;
        }

        //计算最低价

        BigDecimal lowPrice = new BigDecimal(arr[8].trim());
        if (tradeAmtNew.compareTo(lowPrice)==-1){
            lowPrice=tradePriceNew;
        }
        //7.封装结果数据
        SzseAvro szsekAvro = new SzseAvro();
        szsekAvro.setMdStreamID(arr[0].trim());
        szsekAvro.setSecurityID(arr[1].trim());
        szsekAvro.setSymbol(arr[2].trim());
        szsekAvro.setTotalValueTraded(tradeAmtNew.longValue());
        szsekAvro.setTradeVolume(tradeVolTotal);
        szsekAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
        szsekAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
        szsekAvro.setHighPrice(highPrice.doubleValue());
        szsekAvro.setLowPrice(lowPrice.doubleValue());
        szsekAvro.setTradePrice(tradePriceNew.doubleValue());
        szsekAvro.setClosePx(tradePriceNew.doubleValue());
        szsekAvro.setTradingPhaseCode(arr[11].trim());
        szsekAvro.setTimestamp(new Date().getTime());
        szsekAvro.setMarket("szse");

        return szsekAvro;
    }

    private static void parseFile() {
        //1.读取本地文件（深市）

        try {
            final BufferedReader brIndex = new BufferedReader(new InputStreamReader(new FileInputStream("D:\\export\\servers\\tmp\\socket\\szse-index.txt")));
            String line = null;
            while ((line = brIndex.readLine()) != null) {
                final String[] arr = line.split("\\|");
                String code = arr[1].trim();
                final long tradeVol = Long.parseLong(arr[3].trim());
                final long tradeAmt = Double.valueOf(arr[4].trim()).longValue();
                Map<String, Long> volAmtMap = new HashMap<>();
                volAmtMap.put("tradeVol", tradeVol);
                volAmtMap.put("tradeAmt", tradeAmt);
                map.put(code, volAmtMap);
            }
            brIndex.close();

        } catch (Exception e) {
            e.printStackTrace();
        }


        try {
            final BufferedReader brstock = new BufferedReader(new InputStreamReader(new FileInputStream("D:\\export\\servers\\tmp\\socket\\szse-stock.txt")));
            String lineStock = null;
            while ((lineStock = brstock.readLine()) != null) {
                final String[] arr = lineStock.split("\\|");
                final String code = arr[1].trim();
                final long tradeVol = Long.valueOf(arr[3].trim()).longValue();
                final long tradeAmt = Double.valueOf(arr[4].trim()).longValue();
                Map<String, Long> volAmtMap = new HashMap<>();
                volAmtMap.put("tradeVol", tradeVol);
                volAmtMap.put("tradeAmt", tradeAmt);
                map.put(code, volAmtMap);
            }

            brstock.close();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
